// Java
package com.gdhz.xiaohengqin.handler;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gdhz.xiaohengqin.device.constants.WieGandConstant;
import jakarta.annotation.PreDestroy;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
import org.springframework.web.reactive.socket.client.WebSocketClient;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class AccessGateWebSocketHandler implements WebSocketHandler {

    private static final long MAX_RETRIES = Long.MAX_VALUE;
    private static final int INITIAL_BACKOFF_SECONDS = 1;
    private static final int MAX_BACKOFF_MINUTES = 1;
    private static final int BUFFER_SIZE = 1000;
    private static final long TIMEOUT_MS = 45_000;
    private static final long TIMEOUT_SECONDS = 10;

    private final AtomicLong lastMessageTime = new AtomicLong(System.currentTimeMillis());
    private final AtomicBoolean reconnecting = new AtomicBoolean(false);
    private final WebSocketClient client = new ReactorNettyWebSocketClient();
    private final Scheduler messageScheduler;

    private final ObjectMapper objectMapper;
    private final List<MessageHandler> handlers;

    public AccessGateWebSocketHandler(ObjectMapper objectMapper, List<MessageHandler> handlers) {
        this.objectMapper = objectMapper;
        this.handlers = handlers;
        this.messageScheduler = Schedulers.newBoundedElastic(
                Runtime.getRuntime().availableProcessors() * 2,
                1000,
                "MessageProcessor",
                60,
                true
        );
    }

    @Override
    @NonNull
    public Mono<Void> handle(WebSocketSession session) {
        reconnecting.set(false);

        log.info("[{}] WebSocket 连接建立", session.getId());
        lastMessageTime.set(System.currentTimeMillis());

        Flux<WebSocketMessage> heartbeat = Flux.interval(Duration.ofSeconds(TIMEOUT_SECONDS))
                .onBackpressureDrop(t -> log.debug("[{}] 心跳背压丢弃", session.getId()))
                .flatMap(t -> checkConnectionHealth(session))
                .doOnCancel(() -> log.debug("[{}] 心跳发送停止", session.getId()));

        Mono<Void> statusMonitor = Flux.interval(Duration.ofMinutes(1))
                .doOnNext(t -> logStatus(session))
                .onErrorResume(e -> {
                    log.warn("[{}] 状态监控异常: {}", session.getId(), e.getMessage());
                    return Mono.empty();
                })
                .then();

        Mono<Void> messageHandler = session.receive()
                .doOnSubscribe(s -> log.debug("[{}] 开始监听消息", session.getId()))
                .onBackpressureBuffer(
                        BUFFER_SIZE,
                        m -> log.warn("[{}] 消息缓冲区满，丢弃旧消息", session.getId()),
                        BufferOverflowStrategy.DROP_OLDEST
                )
                .map(msg -> {
                    // ✅ 在这里就读取 payload，避免 ByteBuf 被释放
                    String payload = msg.getPayloadAsText();
                    return payload;
                })
                .publishOn(messageScheduler)
                .concatMap(payload -> {
                    lastMessageTime.set(System.currentTimeMillis());
                    return processMessage(payload, session);
                })
                .timeout(Duration.ofMillis(TIMEOUT_MS))
                .then();

        Mono<Void> closeMonitor = session.closeStatus()
                .doOnNext(cs -> {
                    log.warn("[{}] 连接关闭 status={} 准备重连", session.getId(), cs);
                    attemptReconnect("NormalClose", null);
                })
                .then();

        Mono<Void> main = Mono.when(messageHandler, session.send(heartbeat), statusMonitor)
                .doOnTerminate(() -> {
                    if (!session.isOpen()) {
                        log.warn("[{}] 主处理流终止且 session 已关闭", session.getId());
                    } else {
                        log.warn("[{}] 主处理流终止(可能正常完成)", session.getId());
                    }
                })
                .onErrorResume(e -> handleDisconnect(session, e));

        return Mono.firstWithSignal(main, closeMonitor);
    }

    @PreDestroy
    public void cleanup() {
        messageScheduler.dispose();
        log.info("消息处理线程池已关闭");
    }

    private void logStatus(WebSocketSession session) {
        long inactive = System.currentTimeMillis() - lastMessageTime.get();
        log.info("[{}] 状态 - open:{} 空闲:{}ms 内存:{}MB",
                session.getId(),
                session.isOpen(),
                inactive,
                Runtime.getRuntime().totalMemory() / (1024 * 1024));
        if (!session.isOpen()) {
            log.warn("[{}] 巡检发现已关闭(等待已有重连机制处理)", session.getId());
        }
    }

    private Mono<WebSocketMessage> checkConnectionHealth(WebSocketSession session) {
        if (!session.isOpen()) {
            return Mono.error(new IllegalStateException("session closed"));
        }
        long inactive = System.currentTimeMillis() - lastMessageTime.get();
        if (inactive > TIMEOUT_MS) {
            return Mono.error(new IllegalStateException("连接假死(超时未收消息)"));
        }
        return Mono.just(session.pingMessage(dbf -> dbf.wrap("PING".getBytes(StandardCharsets.UTF_8))));
    }

    private Mono<Void> handleDisconnect(WebSocketSession session, Throwable error) {
        log.error("[{}] 连接异常: {}", session.getId(), error.getMessage());
        return Mono.defer(() ->
                        (session.isOpen() ? session.close() : Mono.empty())
                                .onErrorResume(e -> Mono.empty()))
                .doFinally(sig -> attemptReconnect("Error", error))
                .then();
    }

    private Mono<Void> attemptReconnect(String reason, Throwable cause) {
        if (!reconnecting.compareAndSet(false, true)) {
            log.warn("已有重连进行中，跳过。本次原因: {}", reason);
            return Mono.empty();
        }
        log.warn("开始重连, 原因:{} {}", reason, cause != null ? cause.getMessage() : "");
        client.execute(URI.create(WieGandConstant.WEBSOCKET_SERVER_URL), this)
                .retryWhen(Retry.backoff(MAX_RETRIES, Duration.ofSeconds(INITIAL_BACKOFF_SECONDS))
                        .maxBackoff(Duration.ofMinutes(MAX_BACKOFF_MINUTES))
                        .jitter(0.2)
                        .doBeforeRetry(rs -> log.warn("第{}次重连...", rs.totalRetries() + 1)))
                .doOnError(e -> log.error("重连最终失败: {}", e.getMessage()))
                .doFinally(sig -> reconnecting.set(false))
                .subscribe();

        return Mono.empty();
    }

    private Mono<Void> processMessage(String message, WebSocketSession session) {
        log.info("[{}] 收到原始文本: {}", session.getId(), message);
        // ✅ 先过滤掉非 JSON 的“PING”
        if ("PING".equalsIgnoreCase(message.trim())) {
            log.info("[{}] 收到 PING，忽略", session.getId());
            return Mono.empty();
        }
        return Mono.just(message)
                .flatMap(msg -> parseJson(msg, session))
                .flatMap(node -> routeToHandler(node, session))
                .timeout(Duration.ofSeconds(TIMEOUT_SECONDS))
                .onErrorResume(e -> handleMessageError(e, message, session))
                .then();
    }

    private Mono<JsonNode> parseJson(String msg, WebSocketSession session) {
        log.info("[{}] 开始解析 JSON: {}", session.getId(), msg);
        return Mono.fromCallable(() -> objectMapper.readTree(msg))
                .timeout(Duration.ofSeconds(5), Mono.error(new TimeoutException("JSON解析超时")))
                .doOnSuccess(n -> log.info("[{}] 解析成功: {}", session.getId(), n.toPrettyString()))
                .doOnError(e -> log.error("[{}] JSON解析失败: {}", session.getId(), e.getMessage()));
    }

    private Mono<Void> routeToHandler(JsonNode node, WebSocketSession session) {
        return Flux.fromIterable(handlers)
                .filter(h -> h.canHandle(node))
                .next()
                .switchIfEmpty(Mono.error(new NoSuchElementException("无匹配处理器 type=" + node.path("type").asText())))
                .flatMap(h -> h.handle(node))
                .timeout(Duration.ofSeconds(TIMEOUT_SECONDS),
                        Mono.error(new TimeoutException("消息处理超时")));
    }

    private Mono<Void> handleMessageError(Throwable e, String message, WebSocketSession session) {
        log.error("[{}] 消息处理失败: {} 原始: {}", session.getId(), e.getMessage(), message);
        return Mono.empty();
    }
}
